package cn.enjoyedu.rebalance;

import cn.enjoyedu.config.BusiConst;
import cn.enjoyedu.config.KafkaConst;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * 就是一个正常的消费者线程
 * 
 * 特殊的点是, 为了防止业务sql执行了, 偏移量还没提交kafka,本消费者就崩溃了, 
 * 一个事务[业务sql,保存偏移量sql],并且 分区再均衡(包括初始时) 后 seek设置消费者 从数据库指定偏移量开始消费
 */
public class ConsumerWorker implements Runnable {

    private final KafkaConsumer<String, String> consumer;

    //MARK    用来保存每个消费者当前读取分区的偏移量,解决分区再均衡的关键
    private final Map<TopicPartition, OffsetAndMetadata> currOffsetsMap = new HashMap<>();

    private final boolean isStop;
    // 事务类可以送入（tr）
    //private final Transaction  tr  事务类的实例

    public ConsumerWorker(boolean isStop) {
        /*消息消费者配置*/
        Properties properties = KafkaConst.consumerConfig(
                RebalanceConsumer.GROUP_ID,
                StringDeserializer.class,
                StringDeserializer.class);
        // 取消自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        this.isStop = isStop;

        this.consumer = new KafkaConsumer<>(properties);


        //MARK  消费者订阅是加入:  再均衡监听器(HandlerRebalance)
        consumer.subscribe(
                Collections.singletonList(BusiConst.REBALANCE_TOPIC), //主题
                new HandlerRebalance(currOffsetsMap, consumer)//自己写的再均衡监听器, 一个消费者 对 一个监听器
        );
    }

    @Override
    public void run() {

        final String id = Thread.currentThread().getId() + "";
        int count = 0;

        try {

            while (true) {
                
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                
                //事务开始: MARK 精髓就是这2个一起:业务sql,更新偏移量sql
                {
                    for (ConsumerRecord<String, String> record : records) {
                        
                        System.out.println(id + "|" + String.format(
                                "处理主题：%s，分区：%d，偏移量：%d，" +
                                        "key：%s，value：%s",
                                record.topic(),
                                record.partition(),
                                record.offset(),
                                record.key(),
                                record.value()
                        ));


                        // 偏移量 提交到统一HashMap
                        currOffsetsMap.put(
                                new TopicPartition(record.topic(), record.partition()),
                                new OffsetAndMetadata(record.offset() + 1, "no")
                        );
                        count++;
                        //执行业务sql
                    }


                    if (currOffsetsMap.size() > 0) {
                        for (TopicPartition topicPartitionkey : currOffsetsMap.keySet()) {

                            //MARK 每个消费者 消费完成 也会在数据库记录 偏移量
                            //这里不能靠监听器监听, 因为 怕 消费者本身挂了,那它的监听器也没了
                            HandlerRebalance.partitionOffsetDao.put(
                                    topicPartitionkey,
                                    currOffsetsMap.get(topicPartitionkey).offset()
                            );
                        }
                    }
                }
                //事务结束

                //这里是模拟 某消费者崩溃
                if (isStop && count >= 5) {
                    System.out.println(id + "-将关闭，当前偏移量为：" + currOffsetsMap);
                    consumer.commitSync();
                    break;//直接断了,都没机会提交到kafka , 但是没事, 和业务sql同 事务 已经存了偏移量, 我们自己存数据库的偏移量 比kafka里面存的更加准
                }
                
                //提交偏移量到 kafka
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}
